package cn.com.zetatech.loader.common.message;

import cn.com.zetatech.loader.common.constant.LoaderConstant;
import cn.com.zetatech.loader.common.util.EncryptUtil;
import cn.hutool.core.net.NetUtil;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class LogMessageProducer {

    @Value("${HOST_IP:}")
    private String hostIp;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Autowired
    private Gson gson;

    public void jobState(String jobId, String state, String errMessage, String taskInfo, long runDuration) {
        JobStateMessage message = new JobStateMessage();
        message.setJobId(jobId);
        message.setState(state);
        message.setRunDuration(runDuration);
        message.setErrMessage(errMessage);
        message.setPosition(StringUtils.isNotBlank(hostIp) ? hostIp : NetUtil.getLocalhost().getHostAddress());
        message.setTaskInfo(taskInfo);
        kafkaTemplate.send(LoaderConstant.JOB_STATE_MESSAGE, EncryptUtil.encrypt(gson.toJson(message)));
    }

    public void fileState(String jobId,String fileId, String state, String errMessage, String taskInfo, long runDuration) {
        FileStateMessage message = new FileStateMessage();
        message.setEventId(jobId);
        message.setFileId(fileId);
        message.setState(state);
        message.setRunDuration(runDuration);
        message.setErrMessage(errMessage);
        message.setPosition(StringUtils.isNotBlank(hostIp) ? hostIp : NetUtil.getLocalhost().getHostAddress());
        message.setTaskInfo(taskInfo);
        kafkaTemplate.send(LoaderConstant.FILE_STATE_MESSAGE, EncryptUtil.encrypt(gson.toJson(message)));
    }

}
